package com.demo.kafka.consumer;

import com.demo.kafka.entity.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.List;
import java.util.Set;

/**
 * kafka consumer
 * @author K
 */
@Slf4j
@Service
public class KafkaConsumer {

    @Resource(name = "delayConsumer")
    private Consumer<String, KafkaMessage> consumer;

    @Resource(name = "assignConsumer")
    private Consumer<String, KafkaMessage> assignConsumer;

    public Set<TopicPartition> consumer() {
        consumer.subscribe(Set.of("test-topic-1"));
        ConsumerRecords<String, KafkaMessage> records = consumer.poll(Duration.ofMillis(200));
        log.info("消息数量：" + records.count());
        return records.partitions();
    }

    public Set<TopicPartition> consumerPartition1() {
        TopicPartition topicPartition = new TopicPartition("test-topic-1", 1);

        // 用户自己指定消费主体分区，不属于Kafka Group 管理和自动进行rebalance操作
        // group.id必须唯一，否则offset commit 请求将会被拒绝
        // assign + subscribe 同时使用会有冲突错误 java.lang.IllegalStateException: Subscription to topics, partitions and pattern are mutually exclusive
        assignConsumer.assign(List.of(topicPartition));
        ConsumerRecords<String, KafkaMessage> records = assignConsumer.poll(Duration.ofMillis(200));
        log.info("分区1消息数量：" + records.count());
        return records.partitions();
    }
}
